package com.jadows.tdog.message.handler.impl;

import cn.hutool.core.collection.CollUtil;
import com.jadows.tdog.constants.CacheConstants;
import com.jadows.tdog.constants.GlobalActions;
import com.jadows.tdog.domain.ChatMessage;
import com.jadows.tdog.message.WsReqWrapper;
import com.jadows.tdog.message.details.DirectTextMessage;
import com.jadows.tdog.message.handler.IMqMessageHandler;
import com.jadows.tdog.message.handler.pipline.IMessagePipeline;
import com.jadows.tdog.message.handler.pipline.impl.DefaultMessagePipeline;
import com.jadows.tdog.message.handler.pipline.impl.MessageBlockProcess;
import com.jadows.tdog.message.handler.pipline.impl.MessagePersistenceProcess;
import com.jadows.tdog.message.handler.pipline.impl.MessageTicketVerifyProcess;
import com.jadows.tdog.server.codec.JsonWebSocketFrame;
import com.jadows.tdog.service.ContactService;
import com.jadows.tdog.state.OnlineUserRegister;
import com.jadows.tdog.state.SimpleWSSessionUtil;
import com.jadows.tdog.util.UnionUtil;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;

import java.util.Set;

/**
 * 私聊消息处理器
 *
 * @author taojiacheng
 */
@Service
@Slf4j
public class DirectTextMessageHandler implements IMqMessageHandler<WsReqWrapper<DirectTextMessage>> {

    final OnlineUserRegister onlineUserRegister;
    final RedissonClient redissonClient;
    final IMessagePipeline messagePipeline = new DefaultMessagePipeline();
    final ContactService contactService;


    public DirectTextMessageHandler(RedissonClient redissonClient,
                                    ContactService contactService,
                                    OnlineUserRegister onlineUserRegister,
                                    MessageBlockProcess messageBlockProcess,
                                    MessageTicketVerifyProcess messageTicketVerifyProcess,
                                    MessagePersistenceProcess messagePersistenceProcess
    ) {
        // 添加 instanceId, 保证消息只向固定的服务器投递
        RTopic rTopic = redissonClient.getTopic(CacheConstants.DIRECT_TEXT_MSG_TOPIC + onlineUserRegister.getCurrentInstanceId());
        rTopic.addListener(WsReqWrapper.class, (channel, msg) -> {
            @SuppressWarnings("unchecked")
            WsReqWrapper<DirectTextMessage> reqWrapper = (WsReqWrapper<DirectTextMessage>) msg;
            this.arrived(reqWrapper);
        });
        this.onlineUserRegister = onlineUserRegister;
        this.contactService = contactService;
        this.redissonClient = redissonClient;
        // 消息流水线
        this.messagePipeline.addLast(messageBlockProcess) // 黑名单消息处理器
                .addLast(messageTicketVerifyProcess) // 消息票据校验处理器
                .addLast(messagePersistenceProcess) // 消息持久化处理器
        ;
    }

    public ChatMessage handleDirectTextMessage(WsReqWrapper<DirectTextMessage> reqWrapper) {
        DirectTextMessage message = reqWrapper.getData();
        // copy message to chatMessage
        ChatMessage chatMessage = new ChatMessage();
        BeanUtils.copyProperties(message, chatMessage);
        chatMessage.setAction(GlobalActions.DIRECT_TEXT_MSG);
        // unionId
        String unionId = UnionUtil.getUnionId(chatMessage.getFromUid(), chatMessage.getToUid());
        chatMessage.setUnionId(unionId);
        this.messagePipeline.process(chatMessage, reqWrapper.getChannel());
        this.publish(reqWrapper);
        return chatMessage;
    }

    /**
     * 发布消息
     *
     * @param message 消息
     * @return Long
     */
    @Override
    public Long publish(WsReqWrapper<DirectTextMessage> message) {
        DirectTextMessage messageData = message.getData();
        String toUid = messageData.getToUid();
        Set<String> userInstanceIds = onlineUserRegister.getUserInstanceId(toUid);
        // 判断目标用户是否在线
        if (CollUtil.isEmpty(userInstanceIds)) {
            // 不在线
            String fromUid = messageData.getFromUid();
            return contactService.pushMessage(fromUid, toUid, "", messageData.getContent());
        }
        long result = 0L;
        for (String userInstanceId : userInstanceIds) {
            // 推送消息给目标用户
            String topicName = CacheConstants.DIRECT_TEXT_MSG_TOPIC + userInstanceId;
            RTopic rTopic = redissonClient.getTopic(topicName);
            result += rTopic.publish(message);
        }
        return result;
    }


    @Override
    public void arrived(WsReqWrapper<DirectTextMessage> reqWrapper) {
        DirectTextMessage textMessage = reqWrapper.getData();
        String toUid = textMessage.getToUid();
        if (!SimpleWSSessionUtil.isCurrentServer(toUid)) {
            // 不在当前服务器
            log.info("当前服务器不处理目标用户 {} 的消息.", toUid);
            return;
        }
        // 发送消息给目标用户
        Set<Channel> channels = SimpleWSSessionUtil.getChannelByUid(toUid);
        if (CollUtil.isEmpty(channels)) {
            log.warn("用户不在线: {}", toUid);
            return;
        }
        channels.forEach(channel -> channel.writeAndFlush(new JsonWebSocketFrame(reqWrapper)));
    }

}
